package ex.tableapi.join;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.and;

public class ExIntervalJoin extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExIntervalJoin demo = new ExIntervalJoin();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        Table orders1 = demo.registerMysql2("myorder1", "orders1");

        Table result = demo.IntervalJoin(orders);//TODO
        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }

    private Table IntervalJoin(Table orders) {
        Table left = orders.select($("a"), $("b"), $("c"), $("ltime"));
        Table right = orders.select($("d"), $("e"), $("f"), $("rtime"));


        Table result = left.join(right)
                .where(
                        and(
                                $("a").isEqual($("d")),
//                                ltime >=(rtime-5分钟)
//                                $("ltime").isGreaterOrEqual($("rtime")),
//                                $("ltime").isGreaterOrEqual($("rtime").minus(lit(1).minutes()))//minus(lit(1).minutes()) 没试出什么意思
                                $("ltime").isLess($("rtime"))//.plus(lit(10).seconds()))
                        ))
                .select($("a"), $("b"), $("e"), $("rtime"), $("ltime"));
        return result;
    }


    private Table exJoinwithTemporalTable( Table orders) {
//
//// register temporal table function with a time attribute and primary key
//        TemporalTableFunction rates = ratesHistory.createTemporalTableFunction(
//                "r_proctime",
//                "r_currency");
//        tableEnv.registerFunction("rates", rates);
//
//// join with "Orders" based on the time attribute and key
//        Table orders = tableEnv.from("Orders");
//        Table result = orders
//                .joinLateral(call("rates", $("o_proctime")), $("o_currency").isEqual($("r_currency")));
        return null;
    }

    //The UNION operation on two unbounded tables is currently not supported.
    private Table exUnion(Table orders, Table order1) {
//        Table left = tableEnv.from("orders1");
//        Table right = tableEnv.from("orders2");
        Table left = orders.where($("a").isEqual("liji"));
        Table right = order1.where($("a").isEqual("zy"));
        Table union = left.union(right);


        // 下面不能转换，只有流式表可以转成流
        // 出现异常：The UNION operation on two unbounded tables is currently not supported.
        //      DataStream<Tuple2<Boolean, Order>> resultDS = tenv.toRetractStream(result, Order.class);
        //      resultDS.print();

//        left.unionAll(right);
//        left.intersectAll(right);
//        left.intersect(right);
//        left.minus(right);
//        left.minusAll(right);
//        Table result = left.select($("a"), $("b"), $("c")).where($("a").in(right));
        return union;
    }


}
